package studio.raptor.ddal.core.log.service.impl;

import studio.raptor.ddal.core.log.common.CommonMethodUtil;
import studio.raptor.ddal.core.log.common.LogClientConstant;
import studio.raptor.ddal.core.log.queue.IQueueHelper;
import studio.raptor.ddal.core.log.AsyncBatchLog;
import studio.raptor.ddal.core.log.queue.JavaQueue;
import studio.raptor.ddal.core.log.queue.JavaQueueHelper;
import studio.raptor.ddal.core.log.service.interfaces.IThreadManagerService;

import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class LogThreadManagerServiceImpl implements IThreadManagerService {

	private int threadNum= LogClientConstant.threadNum;
    private AsyncBatchLog[] asyncBatchLogs;
	private Future[] futureHandles ;
	private ExecutorService sendMsgService ;

	private String logSwitch=LogClientConstant.logSwitch;
	private String address= LogClientConstant.serverUrl;
	private int timeout=LogClientConstant.timeout;
	private String appName=LogClientConstant.appName;

	private IQueueHelper queueHelper= CommonMethodUtil.getQueueHelper();
    
	public void start() {
		Map<String,JavaQueue> queueMap=queueHelper.getJavaQueueMap();
		//线程数*日志分类(每类日志独立的queue)
		int size=threadNum * queueMap.size();
		asyncBatchLogs = new AsyncBatchLog[size];
		futureHandles = new Future[size];
		sendMsgService = Executors.newFixedThreadPool(size);
		int index=0;
		for(int i = 0 ; i < threadNum ; i ++){
			for (String logType : queueMap.keySet()){
				asyncBatchLogs[index] = new AsyncBatchLog(address, timeout, appName, queueHelper,logType);
				if(!"y".equalsIgnoreCase( logSwitch )){
					asyncBatchLogs[index].setRunFlag(false);
				}
				futureHandles[index] = sendMsgService.submit(asyncBatchLogs[index]);
				index++;
			}
		}
	}

	public void stop() {
		if(asyncBatchLogs.length>0){
			for (int i = 0; i < asyncBatchLogs.length; i++) {
				asyncBatchLogs[i].setRunFlag(false);
			}
		}
		sendMsgService.shutdown();
	}
}